package com.zf.nospring.rocketmq.trans;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.util.HashMap;

/**
 * 事务消息
 */
public class TransProducerTest {
    public static void main(String[] args) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        // 创建事务消息生产者，指定生产者所属的组
        TransactionMQProducer producer = new TransactionMQProducer("no-spring-pg");
        // 指定NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 设置事务监听器
        producer.setTransactionListener(new ProducerTransactionListener());
        // 启动生产者
        producer.start();
        Message message = new Message("test_topic", "事务消息".getBytes());
        // 添加额外参数，可以在事务监听器中获取该参数，也可以不加
        HashMap<String, Object> arg = new HashMap<>();
        arg.put("name", "额外参数");
        // 发送事务消息：先发送半消息、然后执行事务监听器中的本地事务、再根据本地事务的结果决定消息是 可以消费 还是 删除
        TransactionSendResult transactionSendResult = producer.sendMessageInTransaction(message, arg);
        // 消息发出后 sendStatus 就是 ok, 要根据 LocalTransactionState 来判断事务消息是否发送成功
        System.out.printf("发送结果：%s, %s\n", transactionSendResult.getLocalTransactionState(), transactionSendResult.getSendStatus());
        // 关闭生产者
        producer.shutdown();
    }
}
